//! # 死信队列系统
//!
//! 处理消息处理失败的情况，提供重试机制、死信队列管理和告警功能

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use uuid::Uuid;

use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;

/// 死信消息记录
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeadLetterMessage {
    /// 消息ID
    pub id: String,
    /// 原始消息
    pub original_message: Message,
    /// 失败原因
    pub failure_reason: String,
    /// 失败时间
    pub failed_at: chrono::DateTime<chrono::Utc>,
    /// 重试次数
    pub retry_count: u32,
    /// 最后重试时间
    pub last_retry_at: Option<chrono::DateTime<chrono::Utc>>,
    /// 下次重试时间
    pub next_retry_at: Option<chrono::DateTime<chrono::Utc>>,
    /// 消息来源
    pub source_topic: String,
    /// 处理器信息
    pub processor_info: Option<String>,
    /// 错误详情
    pub error_details: HashMap<String, String>,
    /// 元数据
    pub metadata: HashMap<String, String>,
}

/// 死信队列配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeadLetterConfig {
    /// 是否启用死信队列
    pub enabled: bool,
    /// 最大重试次数
    pub max_retry_attempts: u32,
    /// 重试延迟策略
    pub retry_delay_strategy: RetryDelayStrategy,
    /// 死信队列最大消息数
    pub max_dead_letter_messages: usize,
    /// 自动清理过期消息的间隔（秒）
    pub cleanup_interval_secs: u64,
    /// 消息过期时间（秒）
    pub message_ttl_secs: u64,
    /// 是否启用告警
    pub enable_alerts: bool,
    /// 告警阈值配置
    pub alert_thresholds: AlertThresholds,
    /// 是否自动重试
    pub auto_retry: bool,
    /// 重试间隔（秒）
    pub retry_interval_secs: u64,
}

/// 重试延迟策略
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum RetryDelayStrategy {
    /// 固定延迟
    Fixed {
        delay_secs: u64,
    },
    /// 指数退避
    Exponential {
        initial_delay_secs: u64,
        max_delay_secs: u64,
        multiplier: f64,
    },
    /// 线性增长
    Linear {
        initial_delay_secs: u64,
        increment_secs: u64,
    },
}

/// 告警阈值配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AlertThresholds {
    /// 死信消息数量告警阈值
    pub dead_letter_count_threshold: usize,
    /// 错误率告警阈值（百分比）
    pub error_rate_threshold: f64,
    /// 告警检查间隔（秒）
    pub check_interval_secs: u64,
}

/// 死信队列统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeadLetterStats {
    /// 总死信消息数
    pub total_dead_letters: u64,
    /// 待重试消息数
    pub pending_retry_count: u64,
    /// 已重试消息数
    pub retried_count: u64,
    /// 重试成功数
    pub retry_success_count: u64,
    /// 重试失败数
    pub retry_failure_count: u64,
    /// 永久失败消息数
    pub permanent_failure_count: u64,
    /// 平均重试次数
    pub average_retry_count: f64,
    /// 错误率（百分比）
    pub error_rate: f64,
    /// 最近24小时错误数
    pub recent_error_count: u64,
}

/// 死信队列接口
#[async_trait]
pub trait DeadLetterQueue: Send + Sync {
    /// 添加死信消息
    async fn add_dead_letter(
        &self,
        message: Message,
        failure_reason: String,
        error_details: Option<HashMap<String, String>>,
    ) -> MessageBusResult<String>;

    /// 获取待重试的消息
    async fn get_retry_candidates(&self, limit: usize) -> MessageBusResult<Vec<DeadLetterMessage>>;

    /// 重试消息
    async fn retry_message(&self, dead_letter_id: &str) -> MessageBusResult<bool>;

    /// 标记重试成功
    async fn mark_retry_success(&self, dead_letter_id: &str) -> MessageBusResult<()>;

    /// 标记重试失败
    async fn mark_retry_failure(
        &self,
        dead_letter_id: &str,
        failure_reason: String,
    ) -> MessageBusResult<()>;

    /// 获取死信消息
    async fn get_dead_letter(&self, id: &str) -> MessageBusResult<Option<DeadLetterMessage>>;

    /// 删除死信消息
    async fn remove_dead_letter(&self, id: &str) -> MessageBusResult<()>;

    /// 获取统计信息
    async fn get_stats(&self) -> MessageBusResult<DeadLetterStats>;

    /// 清理过期消息
    async fn cleanup_expired(&self) -> MessageBusResult<u64>;
}

/// 告警处理器接口
#[async_trait]
pub trait AlertHandler: Send + Sync {
    /// 发送告警
    async fn send_alert(&self, alert: Alert) -> MessageBusResult<()>;
}

/// 告警信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Alert {
    /// 告警ID
    pub id: String,
    /// 告警级别
    pub level: AlertLevel,
    /// 告警标题
    pub title: String,
    /// 告警消息
    pub message: String,
    /// 告警时间
    pub timestamp: chrono::DateTime<chrono::Utc>,
    /// 相关数据
    pub data: HashMap<String, serde_json::Value>,
}

/// 告警级别
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum AlertLevel {
    /// 信息
    Info,
    /// 警告
    Warning,
    /// 错误
    Error,
    /// 严重错误
    Critical,
}

/// 内存死信队列实现
pub struct InMemoryDeadLetterQueue {
    /// 配置
    config: DeadLetterConfig,
    /// 死信消息存储
    dead_letters: Arc<RwLock<HashMap<String, DeadLetterMessage>>>,
    /// 统计信息
    stats: Arc<RwLock<DeadLetterStats>>,
    /// 告警处理器
    alert_handler: Option<Arc<dyn AlertHandler>>,
}

impl InMemoryDeadLetterQueue {
    /// 创建新的内存死信队列
    pub fn new(config: DeadLetterConfig) -> Self {
        Self {
            config,
            dead_letters: Arc::new(RwLock::new(HashMap::new())),
            stats: Arc::new(RwLock::new(DeadLetterStats::default())),
            alert_handler: None,
        }
    }

    /// 设置告警处理器
    pub fn with_alert_handler(mut self, handler: Arc<dyn AlertHandler>) -> Self {
        self.alert_handler = Some(handler);
        self
    }

    /// 启动后台任务
    pub async fn start(&self) -> MessageBusResult<()> {
        if !self.config.enabled {
            tracing::info!("死信队列已禁用");
            return Ok(());
        }

        tracing::info!("启动死信队列后台任务");

        // 启动自动重试任务
        if self.config.auto_retry {
            self.start_retry_task().await;
        }

        // 启动清理任务
        self.start_cleanup_task().await;

        // 启动告警检查任务
        if self.config.enable_alerts {
            self.start_alert_task().await;
        }

        Ok(())
    }

    /// 启动重试任务
    async fn start_retry_task(&self) {
        let dead_letters = self.dead_letters.clone();
        let stats = self.stats.clone();
        let retry_interval = self.config.retry_interval_secs;

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(retry_interval));
            let mut retry_cycle = 0u64;

            loop {
                interval.tick().await;
                retry_cycle += 1;

                tracing::trace!("执行第 {} 次死信队列重试周期", retry_cycle);

                let retry_candidates = {
                    let dead_letters_guard = dead_letters.read().await;
                    let now = chrono::Utc::now();
                    
                    dead_letters_guard
                        .values()
                        .filter(|dl| {
                            dl.next_retry_at.map_or(false, |retry_time| now >= retry_time)
                        })
                        .cloned()
                        .collect::<Vec<_>>()
                };

                if retry_candidates.is_empty() {
                    tracing::trace!("第 {} 次重试周期: 无待重试消息", retry_cycle);
                    continue;
                }

                tracing::debug!("第 {} 次重试周期: 发现 {} 条待重试消息", retry_cycle, retry_candidates.len());

                for candidate in retry_candidates {
                    // 这里应该触发实际的重试逻辑
                    tracing::debug!("准备重试死信消息: id={}, retry_count={}", 
                        candidate.id, candidate.retry_count);
                    
                    // 更新统计信息
                    let mut stats_guard = stats.write().await;
                    stats_guard.retried_count += 1;
                }
            }
        });
    }

    /// 启动清理任务
    async fn start_cleanup_task(&self) {
        let dead_letters = self.dead_letters.clone();
        let cleanup_interval = self.config.cleanup_interval_secs;
        let message_ttl = self.config.message_ttl_secs;

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(cleanup_interval));

            loop {
                interval.tick().await;
                
                let now = chrono::Utc::now();
                let ttl_duration = chrono::Duration::seconds(message_ttl as i64);
                let mut expired_ids = Vec::new();

                {
                    let dead_letters_guard = dead_letters.read().await;
                    for (id, dead_letter) in dead_letters_guard.iter() {
                        if now.signed_duration_since(dead_letter.failed_at) > ttl_duration {
                            expired_ids.push(id.clone());
                        }
                    }
                }

                if !expired_ids.is_empty() {
                    let mut dead_letters_guard = dead_letters.write().await;
                    for id in &expired_ids {
                        dead_letters_guard.remove(id);
                    }
                    
                    tracing::info!("清理了 {} 条过期死信消息", expired_ids.len());
                }
            }
        });
    }

    /// 启动告警检查任务
    async fn start_alert_task(&self) {
        let dead_letters = self.dead_letters.clone();
        let stats = self.stats.clone();
        let thresholds = self.config.alert_thresholds.clone();
        let alert_handler = self.alert_handler.clone();

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(thresholds.check_interval_secs));

            loop {
                interval.tick().await;

                let dead_letter_count = dead_letters.read().await.len();
                let stats_guard = stats.read().await;

                // 检查死信消息数量告警
                if dead_letter_count > thresholds.dead_letter_count_threshold {
                    if let Some(ref handler) = alert_handler {
                        let alert = Alert {
                            id: Uuid::new_v4().to_string(),
                            level: AlertLevel::Warning,
                            title: "死信队列消息过多".to_string(),
                            message: format!("死信队列中有 {} 条消息，超过阈值 {}", 
                                dead_letter_count, thresholds.dead_letter_count_threshold),
                            timestamp: chrono::Utc::now(),
                            data: {
                                let mut data = HashMap::new();
                                data.insert("count".to_string(), serde_json::Value::Number(dead_letter_count.into()));
                                data.insert("threshold".to_string(), serde_json::Value::Number(thresholds.dead_letter_count_threshold.into()));
                                data
                            },
                        };

                        if let Err(e) = handler.send_alert(alert).await {
                            tracing::error!("发送死信队列告警失败: {}", e);
                        }
                    }
                }

                // 检查错误率告警
                if stats_guard.error_rate > thresholds.error_rate_threshold {
                    if let Some(ref handler) = alert_handler {
                        let alert = Alert {
                            id: Uuid::new_v4().to_string(),
                            level: AlertLevel::Error,
                            title: "错误率过高".to_string(),
                            message: format!("当前错误率 {:.2}%，超过阈值 {:.2}%", 
                                stats_guard.error_rate, thresholds.error_rate_threshold),
                            timestamp: chrono::Utc::now(),
                            data: {
                                let mut data = HashMap::new();
                                data.insert("error_rate".to_string(), serde_json::Value::Number(
                                    serde_json::Number::from_f64(stats_guard.error_rate).unwrap_or_else(|| serde_json::Number::from(0))));
                                data.insert("threshold".to_string(), serde_json::Value::Number(
                                    serde_json::Number::from_f64(thresholds.error_rate_threshold).unwrap_or_else(|| serde_json::Number::from(0))));
                                data
                            },
                        };

                        if let Err(e) = handler.send_alert(alert).await {
                            tracing::error!("发送错误率告警失败: {}", e);
                        }
                    }
                }
            }
        });
    }

    /// 计算下次重试时间
    fn calculate_next_retry_time(&self, retry_count: u32) -> chrono::DateTime<chrono::Utc> {
        let delay = match &self.config.retry_delay_strategy {
            RetryDelayStrategy::Fixed { delay_secs } => {
                Duration::from_secs(*delay_secs)
            }
            RetryDelayStrategy::Exponential { initial_delay_secs, max_delay_secs, multiplier } => {
                let delay = (*initial_delay_secs as f64) * multiplier.powi(retry_count as i32);
                let delay = delay.min(*max_delay_secs as f64) as u64;
                Duration::from_secs(delay)
            }
            RetryDelayStrategy::Linear { initial_delay_secs, increment_secs } => {
                let delay = initial_delay_secs + (increment_secs * retry_count as u64);
                Duration::from_secs(delay)
            }
        };

        chrono::Utc::now() + chrono::Duration::from_std(delay).unwrap_or(chrono::Duration::hours(1))
    }
}

#[async_trait]
impl DeadLetterQueue for InMemoryDeadLetterQueue {
    async fn add_dead_letter(
        &self,
        message: Message,
        failure_reason: String,
        error_details: Option<HashMap<String, String>>,
    ) -> MessageBusResult<String> {
        let dead_letter_id = Uuid::new_v4().to_string();
        let now = chrono::Utc::now();

        tracing::warn!(
            "添加死信消息: id={}, topic={}, reason={}",
            dead_letter_id, message.topic(), failure_reason
        );

        let dead_letter = DeadLetterMessage {
            id: dead_letter_id.clone(),
            original_message: message.clone(),
            failure_reason: failure_reason.clone(),
            failed_at: now,
            retry_count: 0,
            last_retry_at: None,
            next_retry_at: Some(self.calculate_next_retry_time(0)),
            source_topic: message.topic().to_string(),
            processor_info: None,
            error_details: error_details.unwrap_or_default(),
            metadata: HashMap::new(),
        };

        // 检查死信队列大小限制
        {
            let dead_letters_guard = self.dead_letters.read().await;
            if dead_letters_guard.len() >= self.config.max_dead_letter_messages {
                tracing::error!(
                    "死信队列已满: 当前={}, 最大={}",
                    dead_letters_guard.len(),
                    self.config.max_dead_letter_messages
                );
                return Err(MessageBusError::internal_error("死信队列已满"));
            }
        }

        // 添加到死信队列
        {
            let mut dead_letters_guard = self.dead_letters.write().await;
            dead_letters_guard.insert(dead_letter_id.clone(), dead_letter);
        }

        // 更新统计信息
        {
            let mut stats = self.stats.write().await;
            stats.total_dead_letters += 1;
            stats.pending_retry_count += 1;
            
            // 更新错误率
            let total_messages = stats.total_dead_letters + stats.retry_success_count;
            if total_messages > 0 {
                stats.error_rate = (stats.total_dead_letters as f64 / total_messages as f64) * 100.0;
            }
        }

        tracing::info!("死信消息添加成功: id={}", dead_letter_id);
        Ok(dead_letter_id)
    }

    async fn get_retry_candidates(&self, limit: usize) -> MessageBusResult<Vec<DeadLetterMessage>> {
        let dead_letters_guard = self.dead_letters.read().await;
        let now = chrono::Utc::now();

        let candidates: Vec<DeadLetterMessage> = dead_letters_guard
            .values()
            .filter(|dl| {
                dl.retry_count < self.config.max_retry_attempts &&
                dl.next_retry_at.map_or(false, |retry_time| now >= retry_time)
            })
            .take(limit)
            .cloned()
            .collect();

        tracing::debug!("获取重试候选消息: {} 条", candidates.len());
        Ok(candidates)
    }

    async fn retry_message(&self, dead_letter_id: &str) -> MessageBusResult<bool> {
        let mut dead_letters_guard = self.dead_letters.write().await;
        
        if let Some(dead_letter) = dead_letters_guard.get_mut(dead_letter_id) {
            if dead_letter.retry_count >= self.config.max_retry_attempts {
                tracing::warn!(
                    "消息已达到最大重试次数: id={}, retry_count={}",
                    dead_letter_id, dead_letter.retry_count
                );
                return Ok(false);
            }

            dead_letter.retry_count += 1;
            dead_letter.last_retry_at = Some(chrono::Utc::now());
            dead_letter.next_retry_at = Some(self.calculate_next_retry_time(dead_letter.retry_count));

            tracing::info!(
                "开始重试死信消息: id={}, retry_count={}, next_retry={:?}",
                dead_letter_id, dead_letter.retry_count, dead_letter.next_retry_at
            );

            Ok(true)
        } else {
            Err(MessageBusError::not_found("dead_letter", dead_letter_id))
        }
    }

    async fn mark_retry_success(&self, dead_letter_id: &str) -> MessageBusResult<()> {
        let mut dead_letters_guard = self.dead_letters.write().await;
        
        if dead_letters_guard.remove(dead_letter_id).is_some() {
            // 更新统计信息
            let mut stats = self.stats.write().await;
            stats.retry_success_count += 1;
            stats.pending_retry_count = stats.pending_retry_count.saturating_sub(1);
            
            tracing::info!("死信消息重试成功: id={}", dead_letter_id);
            Ok(())
        } else {
            Err(MessageBusError::not_found("dead_letter", dead_letter_id))
        }
    }

    async fn mark_retry_failure(
        &self,
        dead_letter_id: &str,
        failure_reason: String,
    ) -> MessageBusResult<()> {
        let mut dead_letters_guard = self.dead_letters.write().await;
        
        if let Some(dead_letter) = dead_letters_guard.get_mut(dead_letter_id) {
            dead_letter.failure_reason = failure_reason.clone();

            // 更新统计信息
            let mut stats = self.stats.write().await;
            stats.retry_failure_count += 1;

            if dead_letter.retry_count >= self.config.max_retry_attempts {
                // 标记为永久失败
                stats.permanent_failure_count += 1;
                stats.pending_retry_count = stats.pending_retry_count.saturating_sub(1);
                
                tracing::error!(
                    "死信消息永久失败: id={}, retry_count={}, reason={}",
                    dead_letter_id, dead_letter.retry_count, failure_reason
                );
            } else {
                tracing::warn!(
                    "死信消息重试失败: id={}, retry_count={}, reason={}",
                    dead_letter_id, dead_letter.retry_count, failure_reason
                );
            }

            Ok(())
        } else {
            Err(MessageBusError::not_found("dead_letter", dead_letter_id))
        }
    }

    async fn get_dead_letter(&self, id: &str) -> MessageBusResult<Option<DeadLetterMessage>> {
        let dead_letters_guard = self.dead_letters.read().await;
        Ok(dead_letters_guard.get(id).cloned())
    }

    async fn remove_dead_letter(&self, id: &str) -> MessageBusResult<()> {
        let mut dead_letters_guard = self.dead_letters.write().await;
        
        if dead_letters_guard.remove(id).is_some() {
            tracing::info!("删除死信消息: id={}", id);
            Ok(())
        } else {
            Err(MessageBusError::not_found("dead_letter", id))
        }
    }

    async fn get_stats(&self) -> MessageBusResult<DeadLetterStats> {
        let stats = self.stats.read().await.clone();
        Ok(stats)
    }

    async fn cleanup_expired(&self) -> MessageBusResult<u64> {
        let now = chrono::Utc::now();
        let ttl_duration = chrono::Duration::seconds(self.config.message_ttl_secs as i64);
        let mut expired_ids = Vec::new();

        {
            let dead_letters_guard = self.dead_letters.read().await;
            for (id, dead_letter) in dead_letters_guard.iter() {
                if now.signed_duration_since(dead_letter.failed_at) > ttl_duration {
                    expired_ids.push(id.clone());
                }
            }
        }

        let cleaned_count = expired_ids.len() as u64;
        if cleaned_count > 0 {
            let mut dead_letters_guard = self.dead_letters.write().await;
            for id in expired_ids {
                dead_letters_guard.remove(&id);
            }
            
            tracing::info!("清理了 {} 条过期死信消息", cleaned_count);
        }

        Ok(cleaned_count)
    }
}

impl Default for DeadLetterConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            max_retry_attempts: 3,
            retry_delay_strategy: RetryDelayStrategy::Exponential {
                initial_delay_secs: 60,
                max_delay_secs: 3600,
                multiplier: 2.0,
            },
            max_dead_letter_messages: 10000,
            cleanup_interval_secs: 3600,
            message_ttl_secs: 86400 * 7, // 7天
            enable_alerts: true,
            alert_thresholds: AlertThresholds {
                dead_letter_count_threshold: 1000,
                error_rate_threshold: 5.0, // 5%
                check_interval_secs: 300, // 5分钟
            },
            auto_retry: true,
            retry_interval_secs: 300, // 5分钟
        }
    }
}

impl Default for DeadLetterStats {
    fn default() -> Self {
        Self {
            total_dead_letters: 0,
            pending_retry_count: 0,
            retried_count: 0,
            retry_success_count: 0,
            retry_failure_count: 0,
            permanent_failure_count: 0,
            average_retry_count: 0.0,
            error_rate: 0.0,
            recent_error_count: 0,
        }
    }
}